import com.zuikaku.util.TimeUtil;
import org.apache.commons.collections.IteratorUtils;
import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.SocketTextStreamFunction;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

import java.time.Duration;
import java.util.ArrayList;
import java.util.List;

public class WatermarkDemo {
    public static void main(String[] args) {
        //1.创建环境
        StreamExecutionEnvironment environment  = StreamExecutionEnvironment.createLocalEnvironment();
        //2.设置模式与并行度
        environment.setRuntimeMode(RuntimeExecutionMode.AUTOMATIC);
        environment.setParallelism(1);
        //3.指定source，socket
        DataStream<String> dataStream = environment.socketTextStream("139.224.231.81",8888);
        /**
         * 将会输入
         * iPhone12,2023-11:11 00:00:10,1999
         * Redmi12,2023-11:11 00:00:20,999
         * ...
         */
        //4.源数据转换为可以单独处理的字段，用3维元组接收<input,output>
        DataStream<Tuple3<String, String, Integer>> flatMapDs = dataStream.flatMap(new FlatMapFunction<String, Tuple3<String, String, Integer>>() {
            @Override
            public void flatMap(String input, Collector<Tuple3<String, String, Integer>> out) throws Exception {
                String[] inputArr = input.split(",");//[0]itemName,[1]createTime,[2]price
                Tuple3<String, String, Integer> orderTuple = Tuple3.of(inputArr[0], inputArr[1], Integer.parseInt(inputArr[2]));
                out.collect(orderTuple);
            }
        });
        //5.在分组前设定waterMark,允许乱序延迟的最大时间3s
        DataStream<Tuple3<String, String,Integer>> waterMarkDS = flatMapDs.assignTimestampsAndWatermarks(WatermarkStrategy
                //指定最大允许的延迟/乱序 时间
                .<Tuple3<String, String,Integer>>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                .withTimestampAssigner(
                        (event, timestamp) -> {
                            //指定POJO的事件时间列
                            return TimeUtil.strToDate(event.f1).getTime();
                        }
                ));
        //6.按商品名称进行分组
        KeyedStream<Tuple3<String, String, Integer>, String> keyedStream = waterMarkDS.keyBy(new KeySelector<Tuple3<String, String, Integer>, String>() {
            @Override
            public String getKey(Tuple3<String, String, Integer> value) throws Exception {
                return value.f0;
            }
        });
        //用于接收最后兜底的延时数据，侧向输出
        OutputTag<Tuple3<String, String,Integer>> lateData = new OutputTag<Tuple3<String, String,Integer>>("lateData"){};
        //7.进行开窗(时间滚动窗口)每10s统计过去10s的订单综合，并使用全窗口函数进行统计，使用事件时间EventTime
        SingleOutputStreamOperator<String> applied = keyedStream
                .window(TumblingEventTimeWindows.of(Time.seconds(10)))
                .allowedLateness(Time.minutes(1))//允许一分钟内迟到的数据，当迟到数据在他事件时间所属window的endTime之后1min内达到，那么他所在的window将再带上他计算一次
                .sideOutputLateData(lateData)//最后兜底，超过1min未到的延时数据
                .apply(new WindowFunction<Tuple3<String, String, Integer>, String, String, TimeWindow>() {
            @Override
            public void apply(String key, TimeWindow timeWindow, Iterable<Tuple3<String, String, Integer>> iterable, Collector<String> collector) throws Exception {
                //存放窗口中事件时间
                List<String> eventTimeList = new ArrayList<>();
                int total = 0;
                List<Tuple3<String, String, Integer>> orderList = IteratorUtils.toList(iterable.iterator());
                for (Tuple3<String, String, Integer> order : orderList
                ) {
                    total += order.f2;
                    eventTimeList.add(order.f1);
                }
                String result = String.format("分组key:%s,聚合值:%s,窗口开始结束:[%s~%s),窗口所有事件时间:%s",
                        key, total, TimeUtil.format(timeWindow.getStart()), TimeUtil.format(timeWindow.getEnd()), eventTimeList);
                collector.collect(result);
            }
        });
        applied.print("处理后");
        applied.getSideOutput(lateData).printToErr("侧向输出");
        try {
            environment.execute("WaterMark job");
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }
}
